package com.bf.threadPool;


import com.bf.exception.BfException;
import com.bf.utils.CommonService;
import com.bf.utils.SpringUtils;
import lombok.Data;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.support.DefaultTransactionDefinition;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;

/**
 * 多线程操作数据库
 */
@Data
@Slf4j
public class ThreadTransactionalUtils {
    /**
     * 存放子任务
     */
    CopyOnWriteArrayList<CommonService> jobList = new CopyOnWriteArrayList<>();

    /**
     * 表示原始方法jobList是携带了事务
     */
    CopyOnWriteArrayList<CommonService> jobPrimitiveList = new CopyOnWriteArrayList<>();


    /**
     * 存放子任务名称
     */
    CopyOnWriteArrayList<String> jobNameList = new CopyOnWriteArrayList<>();


    /**
     * 任务名称
     */
    private String jobName;

    public ThreadTransactionalUtils(String jobName) {
        this.jobName = jobName;
    }


    @SneakyThrows
    public Boolean runConsumers() {
        log.info("==》无返回，多线程，springboot本地事务，操作数据库开始,任务名称：{}", jobName);
        // 获取线程池
        ExecutorService threadPool = Executors.newFixedThreadPool(5);
        // 每个子线程最大操作时间
        long timeout = 60L;
        // 等待子线程提交或者回滚的时间
        long endTimeout = 120L;
        // 为了让每个线程执行完后回到当前线程，使用CountDownLatch，值为线程数，每次线程执行完就会执行countDown方法减1，为0后回到主线程，也就是当前线程执行后续的代码
        int threadSize = jobList.size();
        CountDownLatch count = new CountDownLatch(threadSize);

        // 用来控制主线程回到子线程
        CountDownLatch mainCount = new CountDownLatch(1);

        // 用来控制最终回到主线程
        CountDownLatch endCount = new CountDownLatch(threadSize);

        // 用来存放子线程的处理结果，若出错就保存一个false
        CopyOnWriteArrayList<Boolean> sonResult = new CopyOnWriteArrayList<Boolean>();

        // 使用线程安全的对象存储，保存主线程最后总的判断结果，是提交还是回滚
        AtomicBoolean ifSubmit = new AtomicBoolean(true);
        // 保存子线程异常
        AtomicReference<Exception> exception = new AtomicReference<Exception>();
        // 添加任务
        for (int i = 0; i < threadSize; i++) {
            ThreadTransactionalUtilsItem thread = new ThreadTransactionalUtilsItem(jobList.get(i), count, sonResult,
                mainCount, ifSubmit, endCount, jobNameList.get(i), exception);
            // 执行方法
            threadPool.execute(thread);
        }

        try {
            // 设置60s超时
            boolean allTasksCompleted = count.await(timeout, TimeUnit.SECONDS);
            if (!allTasksCompleted) {
                log.warn("主线程:任务执行超时！");
                // 处理超时情况，设为回滚
                ifSubmit.set(false);
            } else {
                for (Boolean resp : sonResult) {
                    if (!resp) {
                        log.info("主线程:有线程执行失败，所有线程需要回滚");
                        ifSubmit.set(false);
                        break;
                    }
                }
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            // 回到子线程处理回滚或者提交事务
            mainCount.countDown();
        }
        try {
            // 设置超时，等待子线程处理事务，设为 60 秒
            if (!endCount.await(endTimeout, TimeUnit.SECONDS)) {
                log.warn("主线程:事务提交或回滚超时！");
                ifSubmit.set(false);
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        Exception exception1 = exception.get();
        // 如果是自定义的异常，抛出
        if (!ifSubmit.get() && exception1 instanceof BfException) {
            throw new BfException(exception1.getMessage());
        }
        return ifSubmit.get();
    }

    /**
     * 添加任务
     */
    public ThreadTransactionalUtils addConsumers(CommonService f, String jobName) {
        // 给每个方法添加事务
        PlatformTransactionManager transactionManager = SpringUtils.getBean(PlatformTransactionManager.class);
        jobList.add(() -> {
            // 设置公共属性
            DefaultTransactionDefinition def1 = new DefaultTransactionDefinition();
            def1.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRED);
            // 设置隔离级别为SERIALIZABLE
            def1.setIsolationLevel(TransactionDefinition.ISOLATION_READ_COMMITTED);
            transactionManager.getTransaction(def1);
            f.accept();
        });
        jobPrimitiveList.add(f);
        jobNameList.add(jobName);
        return this;
    }
}
